package stai6271;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.Date;
import java.util.List;

import javax.xml.parsers.ParserConfigurationException;
import javax.xml.parsers.SAXParser;
import javax.xml.parsers.SAXParserFactory;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.thrift.TException;
import org.xml.sax.Attributes;
import org.xml.sax.SAXException;
import org.xml.sax.helpers.DefaultHandler;

/**
 * Modified code from http://javarevisited.blogspot.com/2011/12/parse-read-xml-file-java-sax-parser.html
 *   Currently only prints to console
 * 
 * @author Nel
 *
 */
public class CassQueryParser extends DefaultHandler {
	
	private String temp;
	private String key, LAC, ID, Body, Score, userId;
	private static int keyCount = 0;
	private static Date date = null;
	private static long startLoad = 0;
	private static long endLoad = 0;
	static Cassandra.Client cli = null;
	static private ColumnParent parent = new ColumnParent("users");

	/** The main method sets things up for parsing */
	public static void main(String[] args) throws IOException, SAXException, ParserConfigurationException {
	 	
		//Create a "parser factory" for creating SAX parsers
		SAXParserFactory spfac = SAXParserFactory.newInstance();

		//Now use the parser factory to create a SAXParser object
		SAXParser sp = spfac.newSAXParser();

		//Create an instance of this class; it defines all the handler methods
		CassQueryParser handler = new CassQueryParser();
		
		cli = ClientManager.getConnection();


		try {
			cli.set_keyspace("testbed");
		} catch (InvalidRequestException e) {
			System.out.println("Unable to set keyspace");
			e.printStackTrace();
		} catch (TException e) {
			System.out.println("Unable to set keyspace");
			e.printStackTrace();
		}
		
		startLoad = System.nanoTime();
        //Finally, tell the parser to parse the input and notify the handler
        sp.parse("C:/Users/Nel/Desktop/ask/gets1.xml", handler);
        
        endLoad = System.nanoTime();
        System.out.println("Time to query: " + ((endLoad - startLoad)/1000000));
        System.out.println("Records processed: " + keyCount);
        
      
        ClientManager.closeConnection();
	}

    /**
    * When the parser encounters plain text (not XML elements),
    * it calls(this method, which accumulates them in a string buffer
    */
    public void characters(char[] buffer, int start, int length) {
    	temp = new String(buffer, start, length);
    }
      

       /*
        * Every time the parser encounters the beginning of a new element,
        * it calls this method, which resets the string buffer
        *   This is where we call the program to insert data to the database. 
        */ 
       public void startElement(String uri, String localName,
                     String qName, Attributes attributes) throws SAXException {
              temp = "";
              if (qName.equalsIgnoreCase("row")) {
            	  userId = new String(attributes.getValue("query"));
            	  
            	  ColumnPath path = new ColumnPath("users");
            	  
            	  keyCount++;

                  // read entire row
                  SlicePredicate predicate = new SlicePredicate();
                  SliceRange sliceRange = null;
				try {
					sliceRange = new SliceRange(toByteBuffer(""), toByteBuffer(""), false, 10);
				} catch (UnsupportedEncodingException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
                  predicate.setSlice_range(sliceRange);
                  
                  List<ColumnOrSuperColumn> results = null;
				try {
					results = cli.get_slice(toByteBuffer(userId), parent, predicate, ConsistencyLevel.ONE);
				} catch (UnsupportedEncodingException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (InvalidRequestException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (UnavailableException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (TimedOutException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (TException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
                  for (ColumnOrSuperColumn result : results)
                  {
                      Column column = result.column;
                      try {
						System.out.println(toString(column.name) + " -> " + toString(column.value));
					} catch (UnsupportedEncodingException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
                  }
              }

            	  
            	  
            	  
            	  
            	  
            	  /**            	  //Create user_key_ID
            	  ++keyCount;
            	  
            	  if(keyCount % 100 == 0){
	                	 System.out.println("Count: " + keyCount);
            	  }
            	  
            	  key = new String("bicycle" + keyCount);
            	  
            	  //Get column values
                  LAC =  new String(attributes.getValue("LastActivityDate"));  
                  ID = new String(attributes.getValue("Id"));
                  Body = new String(attributes.getValue("Body"));
                  Score = new String(attributes.getValue("Score"));
            	  
                  try {
					ClientManager.insertRow(cli, key, LAC, ID, Body, Score);
				} catch (UnsupportedEncodingException e) {
					System.out.println("Unable to set column value");
					e.printStackTrace();
				}
              }*/
       }

       /*
        * When the parser encounters the end of an element, it calls this method
        */
       public void endElement(String uri, String localName, String qName)
                     throws SAXException {
           //Do nothing for now            
       }
      
       public static ByteBuffer toByteBuffer(String value) 
       throws UnsupportedEncodingException
       {
           return ByteBuffer.wrap(value.getBytes("UTF-8"));
       }
             
       public static String toString(ByteBuffer buffer) 
       throws UnsupportedEncodingException
       {
           byte[] bytes = new byte[buffer.remaining()];
           buffer.get(bytes);
           return new String(bytes, "UTF-8");
       }
}